package net.bwie.vehicle.dws.function1;

import com.ververica.cdc.connectors.shaded.com.google.common.hash.BloomFilter;
import com.ververica.cdc.connectors.shaded.com.google.common.hash.Funnels;
import net.bwie.vehicle.dws.bean1.VehicleData;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.nio.charset.StandardCharsets;

/**
 * 在线车辆计数器 - 使用ProcessWindowFunction进行去重计数
 * 通过布隆过滤器实现窗口期内车辆VIN码的唯一性判断
 */
public class OnlineVehicleCounterFunction extends
        ProcessWindowFunction<VehicleData, Integer, String, TimeWindow> {

    private transient BloomFilter<CharSequence> bloomFilter;

    @Override
    public void open(Configuration parameters) throws Exception {
     // 初始化BloomFilter实例，用于高效地判断元素是否存在
        bloomFilter = BloomFilter.create(
                Funnels.stringFunnel(StandardCharsets.UTF_8), // 使用UTF-8编码将字符串转换为字节序列
                100000, // 预期插入的元素数量，此处设为10万个
                0.0001 // 误判率，即错误判断元素存在的概率，此处设为0.01%
        );
    }
    @Override
    public void process(String key, Context context,
                        Iterable<VehicleData> elements,
                        Collector<Integer> out) throws Exception {
        // 添加空值检查和初始化验证
        if (bloomFilter == null) {
            open(null);  // 如果bloomFilter未初始化，手动调用open方法
        }
     // 初始化计数器，用于统计新发现的车辆识别号(VIN)数量
        int count = 0;

       // 遍历车辆数据集合，寻找未被布隆过滤器记录的车辆识别号
        for (VehicleData data : elements) {
            // 获取当前车辆数据的车辆识别号
            String vin = data.getVin();
            // 检查车辆识别号是否非空且可能不在布隆过滤器中
            if (vin != null && !bloomFilter.mightContain(vin)) {
                // 将新的车辆识别号添加到布隆过滤器中
                bloomFilter.put(vin);
                // 计数器增加，表示发现了一个新的车辆识别号
                count++;
            }
        }
// 输出新发现的车辆识别号的数量
        out.collect(count);
    }
    @Override
    public void clear(Context context) throws Exception {
        // 清空布隆过滤器，开启下一个窗口
        bloomFilter = null;
    }
}
